package com.example.kafka.service;

import com.example.kafka.config.KafkaConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 用例1 Service
 */
@Service
public class ProducerLeakService {
    private Map<String, KafkaProducer<String, String>> producerMap = new HashMap<>();

    public void send5(String topic, String msg) {
        // 动态参数创建 Producer，未跟踪关闭
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        try {
            producerMap.put(topic, producer);
            producer.send(new ProducerRecord<>(topic, msg));
            // Producer 未关闭，且未做统一管理
        } finally {
            producer.close();
        }
    }

    public void send1(String msg) {
        // 每次调用都 new 一个 Producer，未关闭，导致资源泄露
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        try {
            producer.send(new ProducerRecord<>("topic1", msg));
            // 未调用 producer.close()
        } finally {
            producer.close();
        }
    }

    public void batchSend(List<String> msgs) {
        // 循环内重复创建 Producer，未关闭
        for (String msg : msgs) {
            KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
            try {
                producer.send(new ProducerRecord<>("topic2", msg));
                // 未关闭 producer
            } finally {
                producer.close();
            }
        }
    }

    public void send3(String msg) {
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        try {
            producer.send(new ProducerRecord<>("topic3", msg));
            if (msg == null) {
                throw new RuntimeException("msg is null");
            }
        } finally {
            // 忘记在 finally 里关闭 producer，异常时资源泄露
            producer.close();
        }
    }
}
